package com.loong.kafka.scala.producer

import kafka.javaapi.producer.Producer
import kafka.producer.KeyedMessage
import kafka.producer.ProducerConfig
import java.util.ArrayList
import java.util.Collection
import java.util.List
import java.util.Properties

/**
  * Created by 梁浩峰 on 2016/9/10 14:27.
  */
class LogKafkaProducer {
  var inner: Producer[String, String] =_

  def init(): Producer[String, String] = {
    val properties: Properties = new Properties()
    properties.load(ClassLoader.getSystemResourceAsStream("producer.properties"))
    val config = new ProducerConfig(properties)
    new Producer[String, String](config)
  }

  def send(topicName: String, message: String): Unit = {
    if (topicName == null || message == null) return
    val km: KeyedMessage[String, String] = new KeyedMessage[String, String](topicName, message)
    inner.send(km)
  }

  def send(topicName: String, message: Array[String]): Unit = {
    if (topicName == null || message == null) return
    if (message.isEmpty) return
    val kms =new ArrayList[KeyedMessage[String,String]]()
    for (entry <- message) {
      val km: KeyedMessage[String, String] = new KeyedMessage[String, String](topicName, entry)
      kms.add(km)
    }
    inner.send(kms)
  }

}
object LogKafkaProducer{
  def apply: LogKafkaProducer = new LogKafkaProducer(){
    inner=init()
  }
}
